package org.xxd.kafka.clients.producer.internals;

import org.xxd.kafka.clients.common.util.Time;
import sun.management.Sensor;

import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-09-28 08:51
 * @description: Buffer 的池子，避免频繁的创建太浪费了
 */
public final class BufferPool {
    private final long totalMemory;
    private final long poolableSize;
    private final Deque<ByteBuffer> free;
    private final Deque<Condition> waiters;
    private final ReentrantLock lock;
    private long availableMemory;
    private Time time;

    /**
     * Create a new buffer pool
     *
     * @param memory        The maximum amount of memory that this buffer pool can allocate
     * @param poolableSize  The buffer size to cache in the free list rather than deallocating
     * @param time          time instance
     */
    public BufferPool(long memory, int poolableSize, Time time) {
        this.poolableSize = poolableSize;
        this.lock = new ReentrantLock();
        this.free = new ArrayDeque<ByteBuffer>();
        this.waiters = new ArrayDeque<Condition>();
        this.totalMemory = memory;
        this.availableMemory = memory;
        this.time = time;
    }

    /**
     * 释放内存池
     * @param buffer
     * @param size
     */
    public void deallocate(ByteBuffer buffer, int size) {
        lock.lock();
        try {
            if (size == this.poolableSize && size == buffer.capacity()) {
                buffer.clear();
                this.free.add(buffer);
            } else {
                this.availableMemory += size;
            }
            Condition moreMem = this.waiters.peekFirst();
            if (moreMem != null)
                moreMem.signal();
        } finally {
            lock.unlock();
        }
    }

    public int queued() {
        lock.lock();
        try {
            return this.waiters.size();
        } finally {
            lock.unlock();
        }
    }
}
